package cn.com.tt.flink.sql.engine.strategy;

import cn.com.tt.flink.sql.engine.pojo.source.JobSource;
import cn.com.tt.flink.sql.engine.pojo.source.SourceField;
import cn.com.tt.flink.sql.engine.pojo.source.SourceProperty;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;

public class HbaseTableStrategy implements TableStatementStrategy {
    private JobSource source;

    public HbaseTableStrategy(JobSource source) {
        this.source = source;
    }

    @Override
    public void vaildKeyProperties() throws Exception {
        List<SourceProperty> propertyList = source.getPropertyList();
        PropertyValidator validator = new PropertyValidator(propertyList);
        String tableName = source.getTableName();
        validator.vaild("connector", tableName+"缺少connector属性");
        validator.vaild("table-name", tableName+"缺少table-name属性");
        validator.vaild("zookeeper.quorum", tableName+"缺少zookeeper.quorum属性");
    }

    @Override
    public String createTableStatement() throws Exception {
        String tableName = source.getTableName();
        StringBuilder sb = new StringBuilder();
        //表头信息
        sb.append("create table ").append(tableName).append("(").append("\n");
        //表字段信息
        sb.append(createTableFields());
        //表属性
        sb.append(") with (").append("\n");
        sb.append(createTableProperties());
        sb.append(")");
        return sb.toString();
    }

    public String createTableFields() throws Exception {
        List<SourceField> fieldsList = this.source.getFieldsList();

        Optional<SourceField> primaryField = fieldsList.stream()
                .filter(SourceField::isPrimaryKey)
                .findFirst();

        SourceField pkField = primaryField.orElseThrow(() -> new Exception("缺少primary key配置"));

        StringBuilder sb = new StringBuilder();
        sb.append("`").append(pkField.getFieldName()).append("`").append(" ").append(pkField.getFieldType()).append(",\n");

        sb.append("info row<");
        fieldsList.stream()
                .filter(s -> !s.isPrimaryKey())
                .forEach(field -> {
                    sb.append("`").append(field.getFieldName()).append("`").append(" ").append(field.getFieldType());

                    //有则添加精度信息
                    if (field.getLength() != 0) {
                        sb.append("(").append(field.getLength()).append(",").append(field.getDecimals()).append(")");
                    }

                    sb.append(",");
                });
        //去掉字段拼接的最后一个,
        sb.deleteCharAt(sb.lastIndexOf(","));
        sb.append(">,\n");

        sb.append("primary key(").append(pkField.getFieldName()).append(") not enforced").append("\n");
        return sb.toString();
    }

    public String createTableProperties() {
        Map<String, String> proMap = new HashMap<>();
        //默认属性
        proMap.put("sink.buffer-flush.max-rows", "100");
        proMap.put("sink.buffer-flush.max-size", "4mb");
        proMap.put("sink.buffer-flush.interval", "3s");
        //加载用户属性
        source.getPropertyList().forEach(p -> proMap.put(p.getPropKey(), p.getPropValue()));
        String reduce = proMap.entrySet().stream()
                .map(e -> "'" + e.getKey() + "' = '" + e.getValue() + "',\n")
                .reduce("", (a, b) -> a + b);

        return reduce.substring(0, reduce.lastIndexOf(","));
    }


}
